package com.aotain.jupiter.flink;

import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import com.aotain.common.CommonFunction;

/**
 * 
* @ClassName: SynFloodStat 
* @Description: TODO(这里用一句话描述这个类的作用) 
* @author 程彬
* @date 2017年4月7日 上午11:22:39 
*
 */
public class SynFloodStat implements FlatMapFunction<String, Tuple2<String, String>>{

	/**
	 * serialVersionUID
	 */
	private static final long serialVersionUID = 8188172136381278663L;

	String sourceAreaCountry = null;
	String sourceAreaCity = null;
	String sourceAreaProvince = null;
	String sourceAreaName = null;
	String destAreaCountry = null;   
	String destAreaCity = null;
	String destAreaName = null;
	String destAreaProvince = null;
	String gisstr = null;
	String gisstrBase64 = null;
	String Sourcelon = null;
	String Sourcelat = null;
	String Destlon = null;
	String Destlat = null;
	String sourceAreaId = null;
	String sourceAreaProvinceId = null;
	String sourceAreaCityId = null;
	String destAreaId = null;
	String destAreaProvinceId = null;
	String destAreaCityId = null;
	String areaName_s = null;
	String areaName_d = null;
	String dip = null;
	String sip = null;
	String userAgent = "";
	String cookie = "";
	Long[] IPs = null;
	String dipL = null;
	String sipL = null;
	String dport = null;
	String sport = null;
	String filterIPs = null;
	String domainname = null;
	String url = null;
	String type = null;
	String accesstime = null;
	String idcno = null;
	String delaytime = null;
	Integer status ;
	String refer = "";
	long upStreamOctets = 0;
	long upStreamPacket = 0;
	long dnStreamOctets = 0;
	long dnStreamPacket = 0;
	/**
	 * 
	 * 重载方法
	 * @param in  输入
	 * @param out 输出
	 * @throws Exception
	 */
	@Override
	public void flatMap(String in, Collector<Tuple2<String, String>> out)
			throws Exception {
		// TODO Auto-generated method stub
		
		String logString = in;
		if(logString != null && !"".equals(logString)) {
			String[] splits = logString.split("\\|",-1);
			if(splits.length <= 1)
				return;
			String userAccount = splits[1];
			String ipRex = "([1-9]|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])(\\.(\\d|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])){3}";   
			Pattern pattern = Pattern.compile(ipRex);   
			Matcher matcher = pattern.matcher(userAccount);  
			if(matcher.matches()) {
				if("1".equals(splits[22]) && "6".equals(splits[7]) && splits.length > 42) {//http
					dipL = splits[9];
					dport = splits[10];
					sipL = splits[12];
					sport = splits[13];
					domainname = splits[32];
					url = domainname + "/" + splits[28];
					type = splits[7]; 
					accesstime= splits[14];
					idcno = "911";
					dnStreamOctets = Long.valueOf(splits[17]);
					upStreamOctets = Long.valueOf(splits[18]);
					dnStreamPacket = Long.valueOf(splits[19]);
					upStreamPacket = Long.valueOf(splits[20]);
					userAgent = splits[26];
					cookie = splits[42];
					refer = splits[37];
				}else{//非Http
					dipL = splits[9];
					dport = splits[10];
					sipL = splits[12];
					sport = splits[13];
					domainname = "";
					url = "";
					type = splits[7]; 
					accesstime= splits[14];
					idcno = "911";
					dnStreamOctets = Long.valueOf(splits[17]);
					upStreamOctets = Long.valueOf(splits[18]);
					dnStreamPacket = Long.valueOf(splits[19]);
					upStreamPacket = Long.valueOf(splits[20]);
					status = Integer.valueOf(splits[23]);
				}
				dip = CommonFunction.int2ip(Long.parseLong(dipL));
				sip = CommonFunction.int2ip(Long.parseLong(sipL));
				
				/**
				 * "dip","dport","sip","sport","gisstr","domainname","url","type","delaytime","idcno",
				 * "upstreamoctets","upstreampacket","status","useragent","cookie","post"
				 */
				out.collect(new Tuple2<String,String>(dip,
						String.format("%s_%s_%s_%s_%s_%s_%s_%s_%s_%d_%d_%d_%d_%s_%s_%s",
								dip,dport,sip,sport,domainname,url,type,idcno,upStreamOctets,upStreamPacket,dnStreamOctets,dnStreamPacket,status,userAgent,cookie)));
			}
		}
	}

}
